package jeecg.business.thread;

import java.io.InputStream;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import jeecg.business.entity.enter.EnterEntity;
import jeecg.business.entity.keyword.KeywordEntity;
import jeecg.business.entity.site.SiteEntity;
import jeecg.business.job.FetchTask;
import jeecg.business.service.keyinfo.KeyinfoServiceI;
import jeecg.business.service.log.CollLogServiceI;
import jeecg.business.service.loginuser.LoginUserServiceI;
import jeecg.business.service.site.SimpleSiteServiceI;
import jeecg.business.thread.task.ExecTask;
import jeecg.business.thread.task.impl.BaiduExecTask;
import jeecg.business.thread.task.impl.BaiduVideoExecTask;
import jeecg.business.thread.task.impl.BaiduYinqingExecTask;
import jeecg.business.thread.task.impl.CommonExecTask;
import jeecg.business.thread.task.impl.DongfangCaifuCommonExecTask;
import jeecg.business.thread.task.impl.PeopleExecTask;
import jeecg.business.thread.task.impl.QqVideoExecTask;
import jeecg.business.thread.task.impl.SearchProxyExecTask;
import jeecg.business.thread.task.impl.SinaWeiboExecTask;
import jeecg.business.thread.task.impl.VideoExecTask;
import jeecg.business.thread.task.impl.WeixinExecTask;
import jeecg.business.vo.ProxyIpVo;

import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.conn.params.ConnRoutePNames;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.params.CoreConnectionPNames;
import org.jeecgframework.core.util.ResourceUtil;
import org.jeecgframework.core.util.http.CommonFunc;
import org.jeecgframework.core.util.http.HttpClientUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Repository;

/**
 * 
 * <p>
 * 标题: 抓取管理
 * </p>
 * <p>
 * 描述: // TODO
 * </p>
 * 
 * @author LinYingPiaoJian(Q:71426052)
 * @version // TODO
 * @since 2015-6-6 下午2:43:21
 */
@Repository
@Scope("singleton")
public class HtmlFetchManage {

	static ThreadPoolExecutor execService;
	@Autowired
	KeyinfoServiceI keyinfoService;
	@Autowired
	CollLogServiceI collLogService;
	@Autowired
	SimpleSiteServiceI simpleSiteService;
	@Autowired
	LoginUserServiceI loginUserService;
	

	AtomicInteger execNum;

	Integer maxThread = 30;
	
	
	static AtomicBoolean run = new AtomicBoolean(false);

	
	static ThreadPoolExecutor producerService;
	/**
	 * 生产者-消费者
	 */
	public static int queueSize = Integer.parseInt(ResourceUtil.getSessionattachmenttitle("proxy.uselistnum"));
	
	public static int proxythread = Integer.parseInt(ResourceUtil.getSessionattachmenttitle("proxy.maxThread"));
	
    public static ArrayBlockingQueue<ProxyIpVo> ipqueue = new ArrayBlockingQueue<ProxyIpVo>(queueSize);
    
    static boolean has = false;
    /*
	 * 是否使用代理
	 */
	public static int proxyIsUse = Integer.parseInt(ResourceUtil.getSessionattachmenttitle("proxy.isUse"));
	

    
	public HtmlFetchManage(){
		System.out.println("创建一个抓取管理器");
		
		//如果使用代理
		if(!has && proxyIsUse == 1){
			has = true;
			producerService = (ThreadPoolExecutor) Executors.newFixedThreadPool(proxythread);
			new Thread(){
				public void run() {
					while(true){			//未达到一半时
						try{
							if(ipqueue.size()<=queueSize/2){
								//944798556618652	北京
								//994339998439866	山东
								String ipurl = "http://www.kuaidaili.com/api/getproxy?orderid=998887020668059&an_ha=1&num=100";
								String text = HttpClientUtil.getSourceFromUrl(ipurl, "utf-8", null);
			            		final String [] ips = text.split("\n");
			            		if(ips.length>=50){
				            		for(String ip : ips)
				            			producerService.execute(new IpProducer(ip));
			            		}else{
			            			System.out.println("获取代理地址失败，原因："+text);
			            		}
							}
						}catch(Exception e){
							System.out.println("获取代理地址失败!");
							e.printStackTrace();
						}
						
	            		/**
	            		 * 休眠30秒
	            		 */
	            		try {
							Thread.sleep(30*1000L);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}	
					}
				}
			}.start();
		}
	}
	
	public static ProxyIpVo getIpCanUsed() throws InterruptedException{
		ProxyIpVo p = null;
		Date date = new Date();
		do{
			p = ipqueue.take();
			System.out.println("获取到可用IP：目前还剩下"+ipqueue.size()+"个");
		}while(date.getTime()-p.getTime().getTime()>=10*1000*60L);		//十分钟内有效
		return p;
	}
	/**
	 * 
	 * <p>
	 * 标题: Ip生产者
	 * </p>
	 * <p>
	 * 描述: // TODO
	 * </p>
	 * 
	 * @author LinYingPiaoJian(Q:71426052)
	 * @version // TODO
	 * @since 2015-8-14 下午5:15:54
	 */
	class IpProducer extends Thread{
		
		String ipandport;
		
		public IpProducer(String ipandport){
			this.ipandport = ipandport;
		}
        
        @Override
        public void run() {
            produce();
        }
         
        private void produce() {
			String [] ipport = ipandport.split(":");
			DefaultHttpClient httpclient = new DefaultHttpClient();
			
			
			httpclient.getParams().setParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, HttpClientUtil.timeout); 
	        httpclient.getParams().setParameter(CoreConnectionPNames.SO_TIMEOUT, HttpClientUtil.timeout);
	        
	        //此代理不保证你看到的时候还存活
	        HttpHost proxy = new HttpHost(ipport[0], Integer.parseInt(ipport[1]));
	        httpclient.getParams().setParameter(ConnRoutePNames.DEFAULT_PROXY,proxy);
	        
	        //记得将网址拆成以下形式
	        HttpHost targetHost = new HttpHost("weixin.sogou.com");    //网站名前面不要加http:// 
	        HttpGet httpget = new HttpGet("/weixin?query=%E5%A4%A9%E6%B4%A5%E7%88%86%E7%82%B8&type=2&page=2");
	        
	        try {
				HttpResponse response = httpclient.execute(targetHost, httpget);
				InputStream in = response.getEntity().getContent();
				String searchResultHtml = CommonFunc.uncompressjson(in, "utf-8");
				if(!searchResultHtml.contains("seccode.php")){
					if(searchResultHtml.contains("搜狗")){
						ipqueue.put(new ProxyIpVo(ipandport , new Date()));
						System.out.println("向队列取中插入一个元素("+ipandport+")，队列剩余空间："+(queueSize-ipqueue.size()));
					}
				}else{
//            		System.out.println("可用被拒绝："+ip);
				}
			} catch (Exception e) {
//            	System.out.println("不可用："+ip);
			}finally{
				httpget.releaseConnection();
				httpclient.getConnectionManager().shutdown();
				httpclient.close();
			}
        }
    }
	
	
	
	void setPool(){
		maxThread = Integer.parseInt(ResourceUtil.getSessionattachmenttitle("fetch.maxThread"));
		execService = (ThreadPoolExecutor) Executors.newFixedThreadPool(maxThread);
		execService.setKeepAliveTime(120, TimeUnit.SECONDS);
		execNum = new AtomicInteger(0);
	}

	public void setWork() {
		if(run.get()) return;
		run.set(true);
		setPool();
		new Thread() {
			public void run() {
				System.out.println("开始去轮询。。。");
				while (true){
					int scount = 1;
					int preThreadNum = 0;
					try {
						Thread.sleep(1000L * 10);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					
					if(execService.getActiveCount() == 0 && execService.getTaskCount() == execService.getCompletedTaskCount()){
						System.out.println("活跃连接数："+execService.getActiveCount()+"    重新设置。。");
						execService.shutdownNow();
						FetchTask.runningMap.clear();
						run.set(false);
						execService = null;
						return;
					}else{
						while (execService.getActiveCount() > 0 && execService.getTaskCount() != execService.getCompletedTaskCount()) {
							System.out.println("还在抓取………………………………………………活跃连接数：" + execService.getActiveCount() 
												+ "    任务数：" + execService.getTaskCount() + "   "	
												+ "    完成数：" +execService.getCompletedTaskCount());
							preThreadNum = execService.getActiveCount();
							if (preThreadNum < maxThread){
								scount++;
								System.out.println("运行线程数小于最大线程数（"+preThreadNum+"/"+maxThread+"），第"+scount+"次");
							} else {
								scount = 0;
							}
							execService.allowCoreThreadTimeOut(true);
							if (scount >= 30) {
								execService.shutdownNow();
								FetchTask.runningMap.clear();
								run.set(false);
								execService = null;
								return;
							}
							try {
								Thread.sleep(1000L * 20 * 1);
							} catch (InterruptedException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}
						}
						try {
							Thread.sleep(1000L * 60 * 1);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
					
				}
			}
		}.start();
	}

	public void addTask(EnterEntity enter , KeywordEntity keyword , SiteEntity site) {
		ExecTask task = null;
		if("bdss".equals(site.getId())){
			task = new BaiduYinqingExecTask(enter , keyword, site);
			task.maxpage = (keyword.getSsyqzqys() == null ? 20 : keyword.getSsyqzqys());
		}else if ("ssyq".equals(site.getJclx())) {
			//判断是否使用代理
			if("1".equals(site.getSydl())){
				task = new SearchProxyExecTask(enter , keyword, site);
			}else{
				task = new BaiduExecTask(enter , keyword, site);
			}
			task.maxpage = (keyword.getSsyqzqys() == null ? 20 : keyword.getSsyqzqys());
		} else if ("wx".equals(site.getJclx())) {
			task = new WeixinExecTask(enter , keyword, site);
			task.maxpage = (keyword.getWxzqys()== null ? 5 : keyword.getWxzqys());
//			return;
		} else if ("wb".equals(site.getJclx())) {
			if (site.getZddz().contains("s.weibo.com")) { // 新浪微博
				task = new SinaWeiboExecTask(enter , keyword, site);
				task.maxpage = (keyword.getWbzqys()== null ? 10 : keyword.getWbzqys());
			}
		} else if("dfcfw".equals(site.getId())){		//东方财富网
			task = new DongfangCaifuCommonExecTask(enter, keyword, site);
			task.maxpage = (keyword.getXwmhzqys()== null ? 5 : keyword.getXwmhzqys());
		} else if("people".equals(site.getId())){
			task = new PeopleExecTask(enter, keyword, site);
			task.maxpage = (keyword.getXwmhzqys()== null ? 5 : keyword.getXwmhzqys());
		} else if("qqvideo".equals(site.getId())){		//腾讯视频
			task = new QqVideoExecTask(enter,keyword,site);
			task.maxpage = (keyword.getSpzqys()== null ? 5 : keyword.getSpzqys());
		} else if("baiduvideo".equals(site.getId())){
			task = new BaiduVideoExecTask(enter,keyword,site);
			task.maxpage = (keyword.getSpzqys()== null ? 5 : keyword.getSpzqys());
		}else if("sp".equals(site.getJclx())){		//视频抓取
			task = new VideoExecTask(enter, keyword, site);
			task.maxpage = (keyword.getSpzqys()== null ? 5 : keyword.getSpzqys());
		} else {
			task = new CommonExecTask(enter , keyword, site);
			if("xwmh".equals(site.getJclx()))
				task.maxpage = (keyword.getXwmhzqys()== null ? 5 : keyword.getXwmhzqys());
			if("bk".equals(site.getJclx()))
				task.maxpage = (keyword.getBkzqys()== null ? 5 : keyword.getBkzqys());
			if("lt".equals(site.getJclx()))
				task.maxpage = (keyword.getLtzqys()== null ? 5 : keyword.getLtzqys());
		}
		task.setKeyinfoService(keyinfoService);
		task.setCollLogService(collLogService);
		task.setSimpleSiteService(simpleSiteService);
		task.setLoginUserService(loginUserService);
		execService.execute(task);

	}
}
